#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import os
from base64 import b64decode
import pexpect
import psutil

from common.util.cmd_utils import cmd_format
from redis.common.const import CommandReturnCode, KerberosConstant, RedisCmdRetKey
from redis.common.common import execute_cmd_with_timeout
from redis import log
from common.const import AuthType
from common.common import check_command_injection_ex_quote_space, retry_when_exception
from common.util.check_utils import is_port, is_ip_address
from common.util.check_user_utils import check_path_owner


class RedisClient:
    REDIS_VERSION = 'redis_version'
    REDIS_ROLE = 'role'
    REDIS_SERVER = 'redis-server'
    CLUSTER = 'cluster'
    ROOT_USER = 'root'
    # 10s超时
    EXEC_CMD_TIMEOUT = 10

    @staticmethod
    def query_redis_cluster_process():
        res = []
        processes = psutil.process_iter()
        for process in processes:
            try:
                if RedisClient.REDIS_SERVER not in process.name():
                    continue
                cmd = ' '.join(list(filter(None, process.cmdline())))
                if RedisClient.CLUSTER not in cmd:
                    continue
                tmp = [process.username(), process.exe(), cmd]
                res.append(tmp)
            except Exception as err:
                log.error(f'query redis cluster process error', err)
        return res

    @staticmethod
    def query_redis_cluster_node_info(user, cmd_path, auth, extend_info):
        code, out, err = RedisClient.execute_redis_cmd(user, cmd_path, auth, extend_info, 'info')
        if int(code) != CommandReturnCode.SUCCESS.value:
            log.error(f'execute redis-cli info failed, message: {out}, err: {err}')
            return []
        return out.splitlines()

    @staticmethod
    def query_redis_cluster_all_nodes(user, cmd_path, auth, extend_info):
        code, out, err = RedisClient.execute_redis_cmd(user, cmd_path, auth, extend_info, 'cluster nodes')
        if int(code) != CommandReturnCode.SUCCESS.value:
            log.error(f'execute redis-cli cluster nodes failed, message: {out}, err: {err}')
            return []
        return out.splitlines()

    @staticmethod
    def query_redis_node_dir(user, cmd_path, auth, extend_info):
        code, out, err = RedisClient.execute_redis_cmd(user, cmd_path, auth, extend_info, 'config get dir')
        if int(code) != CommandReturnCode.SUCCESS.value:
            log.error(f'execute redis-cli config get dbfilename failed, message: {out}, err: {err}')
            return ''
        return out.splitlines()[-1]

    @staticmethod
    def query_redis_node_dbfilename(user, cmd_path, auth, extend_info):
        code, out, err = RedisClient.execute_redis_cmd(user, cmd_path, auth, extend_info, 'config get dbfilename')
        if int(code) != CommandReturnCode.SUCCESS.value:
            log.error(f'execute redis-cli config get dbfilename failed, message: {out}, err: {err}')
            return ''
        return out.splitlines()[-1]

    @staticmethod
    def kerberos_auth(user, auth, client_path):
        extend_info = auth.extend_info
        child = None

        try:
            if extend_info[KerberosConstant.CONFIG_MODEL.value] == KerberosConstant.PASSWORD_MODEL.value:
                tmp_cmd = cmd_format("su - {} -c 'export KRB5_CONF={} && source {}/bigdata_env && kinit {}'", user,
                                     extend_info[KerberosConstant.KRB5_CONF.value], client_path, auth.auth_key)
                child = pexpect.spawn(tmp_cmd, encoding='utf-8')
                child.expect(['Password for', pexpect.EOF])
                if RedisCmdRetKey.CLIENT_NOT_FOUND.value in child.before:
                    log.error('kinit failed')
                    raise Exception('kinit failed')
                child.sendline(bytes.decode(b64decode(extend_info[KerberosConstant.SECRET.value])))
                child.expect(pexpect.EOF)
                if RedisCmdRetKey.INFORMATION_ERROR.value in child.before:
                    log.error('password error')
                    raise Exception('password error')
                child.sendline('klist')
                child.expect(pexpect.EOF)
            else:
                tmp_cmd = cmd_format(
                    "su - {} -c 'export KRB5_CONF={} &&source {}/bigdata_env && kinit -kt {} {} && klist'", user,
                    extend_info[KerberosConstant.KRB5_CONF.value], client_path,
                    extend_info[KerberosConstant.KEYTAB.value], auth.auth_key)

                child = pexpect.spawn(tmp_cmd, encoding='utf-8')
                child.expect(pexpect.EOF)
                if RedisCmdRetKey.CLIENT_NOT_FOUND.value in child.before or \
                        RedisCmdRetKey.NO_SUIT_KEY.value in child.before:
                    log.error(f'kinit -kv failed')
                    raise Exception('kinit failed')
            if 'No credentials cache' in child.before:
                log.error('kinit failed')
                raise Exception('kinit failed')
        except Exception as err:
            log.error(f'execute cmd failed', err)
            raise Exception(f'execute cmd failed') from err
        finally:
            child.close()

    @staticmethod
    def gen_cmd_with_auth(auth, redis_conn_cmd, redis_exec_cmd, client_path):
        if auth.auth_type == AuthType.KERBEROS.value:
            return f'source {client_path}/bigdata_env && {redis_conn_cmd} {redis_exec_cmd}'
        elif auth.auth_type == AuthType.NO_AUTO.value:
            return f'source {client_path}/bigdata_env && {redis_conn_cmd} {redis_exec_cmd}'
        else:
            log.error(f'auth type: {auth.auth_type} is not support')
            raise Exception(f'auth type: {auth.auth_type} is not support')

    @staticmethod
    @retry_when_exception(retry_times=3, delay=5)
    def execute_redis_cmd(user, cmd_path, auth, extend_info, redis_exec_cmd):
        if not is_port(extend_info['port']) or not is_ip_address(extend_info['ip']):
            log.error(f"Unexpected ip or port")
            raise Exception(f"Unexpected ip or port")
        client_path = os.path.realpath(extend_info['clientPath'])
        RedisClient.check_client_path(auth, client_path, user)

        auth_type = auth.auth_type
        if auth_type == AuthType.KERBEROS.value:
            RedisClient.kerberos_auth(user, auth, client_path)
        # 判断是否开启ssl加密
        ssl_enable = extend_info.get('sslEnable')
        # 如果没有带ssl开关，执行两次命令：
        if ssl_enable is None or len(ssl_enable) == 0:
            log.info("ssl is empty")
            # 先执行不带ssl的命令，成功直接返回
            redis_conn_cmd = cmd_format("{}/redis-cli -c -h {} -p {}", cmd_path, extend_info['ip'], extend_info['port'])
            code, out, err = RedisClient.execute_redis_full_cmd_with(
                user, auth, redis_conn_cmd, redis_exec_cmd, client_path)
            if int(code) == CommandReturnCode.SUCCESS.value:
                log.info(f'execute cmd: {redis_conn_cmd} {redis_exec_cmd}')
                return code, out, err
            # 执行带ssl的命令，直接返回
            redis_conn_cmd_ssl = cmd_format("{}/redis-cli -c -h {} -p {} --tls", cmd_path, extend_info['ip'],
                                            extend_info['port'])
            log.info(f'execute cmd ssl: {redis_conn_cmd_ssl} {redis_exec_cmd}')
            return RedisClient.execute_redis_full_cmd_with(user, auth, redis_conn_cmd_ssl, redis_exec_cmd, client_path)
        if ssl_enable == 'true':
            redis_conn_cmd = cmd_format("{}/redis-cli -c -h {} -p {} --tls", cmd_path, extend_info['ip'],
                                        extend_info['port'])
        else:
            redis_conn_cmd = cmd_format("{}/redis-cli -c -h {} -p {}", cmd_path, extend_info['ip'], extend_info['port'])

        return RedisClient.execute_redis_full_cmd_with(user, auth, redis_conn_cmd, redis_exec_cmd, client_path)

    @staticmethod
    def check_client_path(auth, client_path, user):
        # 校验环境变量文件是否存在
        if not os.path.exists(os.path.join(client_path, "bigdata_env")):
            log.error(f"{client_path}/bigdata_env not exist")
            raise Exception(f"bigdata_env not exist")
        # 校验环境变量文件用户属主
        if not check_path_owner(os.path.join(client_path, "bigdata_env"), ["root", user]):
            log.error(f"{client_path}/bigdata_env check_path_owner failed, expect root or {user}")
            raise Exception(f"{client_path}/bigdata_env check_path_owner failed, expect root or {user}")
        if check_command_injection_ex_quote_space(auth.auth_key):
            log.error(f"Unexpected input")
            raise Exception(f"Unexpected input")

    @staticmethod
    def execute_redis_full_cmd_with(user, auth, redis_conn_cmd, redis_exec_cmd, client_path):
        dst_cmd = RedisClient.gen_cmd_with_auth(auth, redis_conn_cmd, redis_exec_cmd, client_path)
        log.info(f'execute cmd: {redis_conn_cmd} {redis_exec_cmd}')
        if user == RedisClient.ROOT_USER:
            return execute_cmd_with_timeout(f"su -c '{dst_cmd}'", RedisClient.EXEC_CMD_TIMEOUT)
        else:
            return execute_cmd_with_timeout(f"su - {user} -c '{dst_cmd}'", RedisClient.EXEC_CMD_TIMEOUT)
